package com.atguigu.gamll.realtime.app.dws;

import com.atguigu.gamll.realtime.app.dwd.BaseApp;
import com.atguigu.gamll.realtime.beans.ProvinceStats;
import com.atguigu.gamll.realtime.utils.ClickHouseUtil;
import com.atguigu.gamll.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// 地区主题统计
public class ProvinceStatsApp extends BaseApp {
    public static void main(String[] args) throws Exception {
        ProvinceStatsApp provinceStatsApp = new ProvinceStatsApp();
        provinceStatsApp.baseEntry();
    }

    @Override
    public void biz(StreamExecutionEnvironment env) {
        // TODO 1.指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2.从Kafka主题中读取数据 创建动态表
        String topic = "dwm_order_wide";
        String groupId = "province_stats_group";
        tableEnv.executeSql("CREATE TABLE order_wide (" +
                "province_id BIGINT," +
                "province_name STRING," +
                "province_area_code STRING," +
                "province_iso_code STRING," +
                "province_3166_2_code STRING," +
                "order_id BIGINT," +
                "split_total_amount DECIMAL(16, 2)," +
                "create_time STRING," +
                "rowtime AS TO_TIMESTAMP(create_time)," +
                "WATERMARK FOR rowtime AS rowtime - INTERVAL '3' SECOND" +
                ") WITH (" + MyKafkaUtil.getKafkaDDL(topic, groupId) + ")" );

        // TODO 3.从动态表中查询数据
        Table tableResult = tableEnv.sqlQuery("select " +
                "DATE_FORMAT(TUMBLE_START(rowtime, interval '10' second), 'yyyy-MM-dd HH:mm:ss') stt," +
                "DATE_FORMAT(TUMBLE_END(rowtime, interval '10' second), 'yyyy-MM-dd HH:mm:ss')  edt," +
                "province_id," +
                "province_name," +
                "province_area_code area_code," +
                "province_iso_code iso_code," +
                "province_3166_2_code iso_3166_2," +
                "count(distinct order_id) order_count," +
                "sum(split_total_amount) order_amount," +
                "UNIX_TIMESTAMP()*1000 ts " +
                "from order_wide " +
                "group by TUMBLE(rowtime, interval '10' second)," +
                "province_id,province_name,province_area_code,province_iso_code,province_3166_2_code");

        // TODO 4.将动态表转换为流
        DataStream<ProvinceStats> provinceStatsDS = tableEnv.toAppendStream(tableResult, ProvinceStats.class);

        // TODO 5.将流中的数据写入clickhouse
        provinceStatsDS.print(">>>>");
        provinceStatsDS.addSink(
                ClickHouseUtil.getSinkFunction("insert into  province_stats_0819 values(?,?,?,?,?,?,?,?,?,?)")
        );
    }
}
